package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"6.824/labgob"
	"bytes"
	"fmt"
	"math"
	"math/rand"
	//	"bytes"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"
	"6.824/labrpc"
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type Status int

const (
	Follower Status = iota
	Candidate
	Leader
)

// A Go object implementing a single Raft peer.
type Raft struct {
	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	mu                sync.Mutex          // Lock to protect shared access to this peer's state
	peers             []*labrpc.ClientEnd // RPC end points of all peers
	persister         *Persister          // Object to hold this peer's persisted state
	me                int                 // this peer's index into peers[]
	dead              int32               // set by Kill()
	applyCond         *sync.Cond          // apply cond
	applyCh           *chan ApplyMsg
	currentTerm       int
	voteFor           int
	commitIndex       int // 已提交的日志的索引，日志索引从 1 计数
	lastApplied       int // 最后被应用到状态机日志的索引，初始为 0
	logs              []LogEntry
	nextIndex         []int // 初始值为领导人最后的日志条目 + 1
	matchIndex        []int
	status            Status
	flushTime         int64
	LastIncludedIndex int // 快照中最后一个日志的索引，初始为 0
	LastIncludedTerm  int
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	var term int = rf.currentTerm
	var isleader bool = rf.status == Leader
	return term, isleader
}

func (rf *Raft) persistData() []byte {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.voteFor)
	e.Encode(rf.logs)
	e.Encode(rf.LastIncludedIndex)
	e.Encode(rf.LastIncludedTerm)
	data := w.Bytes()
	return data
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	rf.persister.SaveRaftState(rf.persistData())
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var voteFor int
	var lastIncludedIndex int
	var lastIncludedTerm int
	var logs []LogEntry

	if d.Decode(&currentTerm) != nil ||
		d.Decode(&voteFor) != nil ||
		d.Decode(&logs) != nil ||
		d.Decode(&lastIncludedIndex) != nil ||
		d.Decode(&lastIncludedTerm) != nil {
		DPrintf("decode error")
	} else {
		rf.currentTerm = currentTerm
		rf.voteFor = voteFor
		rf.logs = logs
		rf.LastIncludedIndex = lastIncludedIndex
		rf.LastIncludedTerm = lastIncludedTerm
		rf.lastApplied = lastIncludedIndex
		rf.commitIndex = lastIncludedIndex
	}
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {
	// Your code here (2D).
	return true
}

// 根据日志索引获取日志在数组中的下标
func (rf *Raft) getArrayIndexByLogIndex(logIndex int) int {
	return logIndex - rf.LastIncludedIndex - 1
}

// 获取最后一个日志的索引
func (rf *Raft) getLastLogIndex() int {
	return len(rf.logs) + rf.LastIncludedIndex
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if index <= rf.LastIncludedIndex || index > rf.commitIndex {
		return
	}

	subLog := rf.logs[rf.getArrayIndexByLogIndex(index)+1:]

	rf.LastIncludedTerm = rf.logs[rf.getArrayIndexByLogIndex(index)].Term
	rf.LastIncludedIndex = index
	rf.logs = subLog
	// 快照也相当与是提交
	if index > rf.lastApplied {
		rf.lastApplied = index
	}
	rf.persister.SaveStateAndSnapshot(rf.persistData(), snapshot)

	DPrintf("%s start snapshot, index is %d, now lastIncludedIndex is %d, commitIndex is %d, log %+v\n", rf.Self(), index, rf.LastIncludedIndex, rf.commitIndex, rf.logs)
}

type LogEntry struct {
	Command interface{}
	Term    int
}

type AppendEntriesArgs struct {
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []LogEntry
	Term         int
	LeaderCommit int
}

type AppendEntriesReply struct {
	Success       bool
	Term          int
	ConflictIndex int // 仅在日志冲突时使用，其余时候为 0
	ConflictTerm  int // 仅在日志冲突时使用，其余时候为 0
}

type InstallSnapshotArgs struct {
	Term             int
	LeaderId         int
	LastIncludeIndex int
	LastIncludeTerm  int
	Data             []byte
}

type InstallSnapshotReply struct {
	Term int
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term    int
	Success bool
}

// AppendEntries 接受来自领导人的日志同步请求或心跳
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	//DPrintf("server %d.%d.%d reciver leader append entries request with args %+v\n", rf.me, rf.status, rf.currentTerm, args)

	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()

	if args.Term < rf.currentTerm {
		reply.Success, reply.Term = false, rf.currentTerm
		return
	}

	rf.status = Follower
	rf.currentTerm = args.Term
	rf.voteFor = args.LeaderId
	rf.flushTime = time.Now().UnixMilli()

	//DPrintf("%s reciver leader %d message %+v and flush time is %d\n", rf.self(), args.LeaderId, args, rf.flushTime)

	reply.Term = rf.currentTerm

	if args.PrevLogIndex > rf.LastIncludedIndex {
		// 日志不匹配，自身没有 PrevLog，要求 Leader 发送自身最后一个日志的下一个日志
		if rf.getLastLogIndex() < args.PrevLogIndex {
			reply.Success, reply.ConflictIndex = false, rf.getLastLogIndex()+1
			return
		}

		// 否则，比较 term，如果 term 不匹配，则保存 ConflictTerm，并找到 ConflictTerm 最早出现的下标 ConflictIndex
		// 如果 Leader 日志中没有 ConflictTerm，则 Follower 需要删除从 ConflictIndex 开始之后的日志
		// 如果 Leader 存在，则从 Leader 最后一个 term 为 ConflictTerm 的日志开始同步
		if rf.logs[rf.getArrayIndexByLogIndex(args.PrevLogIndex)].Term != args.PrevLogTerm {
			reply.Success, reply.ConflictTerm = false, rf.logs[rf.getArrayIndexByLogIndex(args.PrevLogIndex)].Term
			for index := rf.LastIncludedIndex + 1; index <= args.PrevLogIndex; index++ {
				if rf.logs[rf.getArrayIndexByLogIndex(index)].Term == reply.ConflictTerm {
					reply.ConflictIndex = index
					break
				}
			}
			return
		}
	}

	if len(args.Entries) != 0 {
		// 找到第一个不匹配的位置，并且覆盖后续日志
		for index := args.PrevLogIndex + 1; index <= args.PrevLogIndex+len(args.Entries); index++ {
			i := index - args.PrevLogIndex - 1
			if index <= rf.LastIncludedIndex {
				continue
			}
			if index > rf.getLastLogIndex() || rf.logs[rf.getArrayIndexByLogIndex(index)].Term != args.Entries[i].Term {
				rf.logs = append(rf.logs[0:rf.getArrayIndexByLogIndex(index)], args.Entries[i:]...)
			}
		}
		//rf.logs = append(rf.logs[0:rf.getArrayIndexByLogIndex(args.PrevLogIndex)+1], args.Entries...)
		DPrintf("%s reciver leader append entries request %+v and ready to accpect, "+
			"log length is %d, commitIndex is %d, lastInclude is %d. log %+v\n", rf.Self(), args, len(rf.logs), rf.commitIndex, rf.LastIncludedIndex, rf.logs)
	}

	if args.LeaderCommit > rf.commitIndex && args.LeaderCommit <= rf.getLastLogIndex() {
		rf.commitIndex = int(math.Min(float64(args.LeaderCommit), float64(rf.getLastLogIndex())))
		rf.applyCond.Signal()
	}
	reply.Success = true
}

// RequestVote
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()

	//DPrintf("server %d.%d.%d reciver the vote request with args %+v\n", rf.me, rf.status, rf.currentTerm, args)

	if args.Term < rf.currentTerm || (args.Term == rf.currentTerm && rf.voteFor != args.CandidateId) || rf.status == Leader {
		reply.Success = false
		reply.Term = rf.currentTerm
		return
	}

	myLastLogIndex := rf.getLastLogIndex()
	myLastLogTerm := rf.LastIncludedTerm
	if myLastLogIndex > rf.LastIncludedIndex {
		myLastLogTerm = rf.logs[rf.getArrayIndexByLogIndex(myLastLogIndex)].Term
	}

	if args.LastLogTerm < myLastLogTerm || (args.LastLogIndex < myLastLogIndex && args.LastLogTerm == myLastLogTerm) {
		reply.Success = false
		reply.Term = rf.currentTerm
		return
	}

	rf.status = Follower
	rf.currentTerm = args.Term
	rf.voteFor = args.CandidateId
	rf.flushTime = time.Now().UnixMilli()

	reply.Success = true
	reply.Term = rf.currentTerm
}

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		return
	}
	rf.status = Follower
	rf.currentTerm = args.Term
	rf.voteFor = args.LeaderId
	rf.flushTime = time.Now().UnixMilli()
	rf.persist()
	reply.Term = rf.currentTerm

	if args.LastIncludeIndex <= rf.LastIncludedIndex || args.LastIncludeIndex <= rf.commitIndex {
		return
	}
	// 丢弃快照之前的日志
	subLog := make([]LogEntry, 0)
	for i := args.LastIncludeIndex + 1; i <= rf.getLastLogIndex(); i++ {
		subLog = append(subLog, rf.logs[rf.getArrayIndexByLogIndex(i)])
	}
	//subLog := rf.logs[rf.getArrayIndexByLogIndex(args.LastIncludeIndex)+1:]  args.LastIncludeIndex 可能超出数组长度，不能直接用切片
	rf.LastIncludedIndex = args.LastIncludeIndex
	rf.LastIncludedTerm = args.LastIncludeTerm
	rf.logs = subLog
	if rf.LastIncludedIndex > rf.commitIndex {
		rf.commitIndex = rf.LastIncludedIndex
	}
	if rf.LastIncludedIndex > rf.lastApplied {
		rf.lastApplied = rf.LastIncludedIndex
	}

	rf.persister.SaveStateAndSnapshot(rf.persistData(), args.Data)
	DPrintf("%s get leader snapshot, now lastIndex is %d, log is %+v\n", rf.Self(), rf.LastIncludedIndex, rf.logs)
	go func() {
		*rf.applyCh <- ApplyMsg{
			SnapshotValid: true,
			Snapshot:      args.Data,
			SnapshotIndex: args.LastIncludeIndex,
			SnapshotTerm:  args.LastIncludeTerm,
		}
	}()
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

// sendAppendEntries 发送 RPC 请求到服务方
func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

// sendSnapShot 发送 RPC 请求到服务方
func (rf *Raft) sendSnapShot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	ok := rf.peers[server].Call("Raft.InstallSnapshot", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := false

	// Your code here (2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()

	if rf.killed() || rf.status != Leader {
		return index, term, isLeader
	}

	index = rf.getLastLogIndex() + 1
	term = rf.currentTerm
	rf.logs = append(rf.logs, LogEntry{Term: term, Command: command})

	DPrintf("%s reciver log %+v, the logs length is %d\n", rf.Self(), command, len(rf.logs))

	rf.syncLog()
	return index, term, true
}

func (rf *Raft) syncLogSnapshot(id int, prevTerm int) {
	ok := false
	for ok == false && rf.killed() == false {
		rf.mu.Lock()
		// 身份过期，则退出，避免同步错误的日志，因此这里一定需要锁
		if rf.status != Leader || rf.currentTerm != prevTerm || rf.nextIndex[id] > rf.LastIncludedIndex {
			rf.mu.Unlock()
			return
		}

		args := InstallSnapshotArgs{
			Term:             rf.currentTerm,
			LeaderId:         rf.me,
			LastIncludeIndex: rf.LastIncludedIndex,
			LastIncludeTerm:  rf.LastIncludedTerm,
			Data:             rf.persister.ReadSnapshot(),
		}
		reply := InstallSnapshotReply{}

		rf.mu.Unlock() // 避免网络堵塞导致锁长时间无法释放
		ok = rf.sendSnapShot(id, &args, &reply)

		if ok {
			DPrintf("%s send snapshot to server %d and get reply %+v", rf.Self(), id, reply)
			rf.mu.Lock()
			if reply.Term > rf.currentTerm {
				rf.flushTime = time.Now().UnixMilli()
				rf.status = Follower
				rf.voteFor = -1
				rf.currentTerm = reply.Term
				rf.persist()
				rf.mu.Unlock()
				return
			}

			rf.matchIndex[id] = args.LastIncludeIndex
			rf.nextIndex[id] = rf.matchIndex[id] + 1
			DPrintf("%s update matchIndex and nextIndex %+v, %+v\n", rf.Self(), rf.matchIndex, rf.nextIndex)
			rf.mu.Unlock()
		}
	}
}

// syncLog 开始向其他节点同步日志，此方法启动协程发起同步，主方法不会堵塞
func (rf *Raft) syncLog() {
	DPrintf("%s start sync log to followers, last index %d\n", rf.Self(), len(rf.logs))

	if rf.commitIndex == rf.getLastLogIndex() {
		return
	}
	prevTerm := rf.currentTerm

	for i := range rf.peers {
		if i == rf.me || rf.matchIndex[i] == rf.getLastLogIndex() {
			continue
		}
		go func(id int) {

			for rf.status == Leader && rf.matchIndex[id] != rf.getLastLogIndex() {
				rf.mu.Lock()
				// 身份过期，则退出，避免同步错误的日志，因此这里一定需要锁
				if rf.status != Leader || rf.currentTerm != prevTerm {
					rf.mu.Unlock()
					return
				}

				if rf.nextIndex[id] <= rf.LastIncludedIndex {
					rf.mu.Unlock()
					rf.syncLogSnapshot(id, prevTerm)
					time.Sleep(10 * time.Millisecond)
					continue
				}

				args := AppendEntriesArgs{
					Term:         rf.currentTerm,
					LeaderId:     rf.me,
					LeaderCommit: int(math.Min(float64(rf.commitIndex), float64(rf.matchIndex[id]))),
					PrevLogIndex: rf.nextIndex[id] - 1,
					PrevLogTerm:  rf.LastIncludedTerm,
					Entries:      rf.logs[rf.getArrayIndexByLogIndex(rf.nextIndex[id]):],
				}
				if args.PrevLogIndex > rf.LastIncludedIndex {
					args.PrevLogTerm = rf.logs[rf.getArrayIndexByLogIndex(args.PrevLogIndex)].Term
				}
				rf.mu.Unlock()

				reply := AppendEntriesReply{}
				ok := rf.sendAppendEntries(id, &args, &reply)

				if !ok {
					time.Sleep(10 * time.Millisecond)
					continue
				}

				rf.mu.Lock()

				if rf.status != Leader || rf.currentTerm != prevTerm {
					rf.mu.Unlock()
					return
				}

				DPrintf("%s sync log %+v to %d and get reply %+v\n", rf.Self(), args, id, reply)

				if reply.Term > rf.currentTerm {
					rf.flushTime = time.Now().UnixMilli()
					rf.status = Follower
					rf.voteFor = -1
					rf.currentTerm = reply.Term
					rf.persist()
					rf.mu.Unlock()
					return
				}

				if reply.Success {
					rf.matchIndex[id] = args.PrevLogIndex + len(args.Entries)
					rf.nextIndex[id] = rf.matchIndex[id] + 1
					DPrintf("%s reciver server %d sync log approve, now the match and next index is %+v, %+v, lastIndex is %d\n", rf.Self(), id, rf.matchIndex, rf.nextIndex, rf.LastIncludedIndex)
				} else if reply.ConflictIndex > 0 || reply.ConflictTerm > 0 {
					// 发生冲突，更新 nextIndex
					if reply.ConflictTerm <= 0 {
						rf.nextIndex[id] = reply.ConflictIndex
					} else {
						// Leader 在 PrevIndex 处的任期不匹配
						// 如果 Leader 日志中没有 ConflictTerm，则 Follower 的所有 ConflictTerm 日志都需要删除，nextIndex 刚好是 ConflictIndex
						// 否则，可能会存在部分匹配，则从最后一个开始尝试起
						ci := reply.ConflictIndex
						for index := args.PrevLogIndex; index > rf.LastIncludedIndex; index-- {
							if rf.logs[rf.getArrayIndexByLogIndex(index)].Term == reply.ConflictTerm {
								ci = index
								break
							}

							if rf.logs[rf.getArrayIndexByLogIndex(index)].Term < reply.ConflictTerm {
								break
							}
						}
						rf.nextIndex[id] = ci
					}
					DPrintf("%s update server %d next index %d, self log length %d\n", rf.Self(), id, rf.nextIndex[id], len(rf.logs))
				}
				// 惰性跟进 commitIndex
				for c := rf.getLastLogIndex(); c > rf.commitIndex; c-- {
					if rf.getArrayIndexByLogIndex(c) < 0 {
						DPrintf("%s, c is %d, commitIndex %d, lastIncludeIndex %d, lastApplied %d, matchIndex %+v,log %+v",
							rf.Self(), c, rf.commitIndex, rf.LastIncludedIndex, rf.lastApplied, rf.matchIndex, rf.logs)
					}
					if rf.logs[rf.getArrayIndexByLogIndex(c)].Term != rf.currentTerm {
						break
					}
					num := 1
					for j := range rf.matchIndex {
						if j == rf.me {
							continue
						}

						if rf.matchIndex[j] >= c {
							num++
						}
					}

					// 存在一个索引 c，使得大多数的 match[index] >= c 且对应日志的 term 等于当前 term，则推进 commitIndex
					if num > len(rf.peers)/2 {
						rf.commitIndex = c
						rf.applyCond.Signal()
						break
					}
				}
				rf.mu.Unlock()
				time.Sleep(10 * time.Millisecond)
			}
		}(i)
	}
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) electionTicker() {
	var lastFlushTime int64 = -1

	for rf.killed() == false {
		flushTime := rf.flushTime

		rand.Seed(time.Now().UnixMilli() + int64(rf.me)*2023)
		overtime := int64(200 + rand.Intn(400))

		if flushTime != lastFlushTime {
			overtime = overtime + flushTime - time.Now().UnixMilli()
		}
		lastFlushTime = flushTime

		if overtime > 0 {
			time.Sleep(time.Duration(overtime) * time.Millisecond)
		}

		rf.mu.Lock()
		// 休眠期间没有刷新则确定为超时
		if rf.flushTime == flushTime && rf.status != Leader {
			//DPrintf("server %d-%v-%d election ticker on time %v\n", rf.me, rf.status, rf.currentTerm, time.Now().UnixMilli())
			rf.doElection()
		}
		rf.mu.Unlock()
	}
}

func (rf *Raft) heartbeatTicker() {
	for rf.killed() == false {
		time.Sleep(50 * time.Millisecond)
		if rf.status == Leader {
			rf.doHeartbeat()
		}
	}
}

// doElection 方法开启协程向其他节点发送投票请求，一旦超过半数以上节点同意，则转化为 Leader
func (rf *Raft) doElection() {
	var prevTerm = rf.currentTerm
	// 更新状态
	rf.status = Candidate
	rf.voteFor = rf.me
	rf.currentTerm++
	rf.persist()

	var voteNum int32 = 1
	var total int32 = int32(len(rf.peers))

	for i := range rf.peers {
		if i == rf.me {
			continue
		}
		go func(id int) {
			for !rf.killed() && rf.currentTerm == prevTerm+1 && rf.status == Candidate {
				rf.mu.Lock()
				if rf.currentTerm != prevTerm+1 || rf.status != Candidate {
					rf.mu.Unlock()
					return
				}
				lastLogIndex := rf.getLastLogIndex()
				lastLogTerm := rf.LastIncludedTerm
				if lastLogIndex > rf.LastIncludedIndex {
					lastLogTerm = rf.logs[rf.getArrayIndexByLogIndex(lastLogIndex)].Term
				}
				args := RequestVoteArgs{
					Term:         rf.currentTerm,
					CandidateId:  rf.me,
					LastLogIndex: lastLogIndex,
					LastLogTerm:  lastLogTerm,
				}
				rf.mu.Unlock()
				reply := RequestVoteReply{}
				ok := rf.sendRequestVote(id, &args, &reply)

				rf.mu.Lock()
				if rf.currentTerm != prevTerm+1 || rf.status != Candidate {
					rf.mu.Unlock()
					return
				}

				if ok {
					DPrintf("%s send vote request %+v to server %d and get reply %+v\n", rf.Self(), args, id, reply)
					if reply.Term > rf.currentTerm {
						rf.flushTime = time.Now().UnixMilli()
						rf.status = Follower
						rf.voteFor = -1
						rf.currentTerm = reply.Term
						rf.persist()
						rf.mu.Unlock()
						return
					}
					if reply.Success {
						voteNum++
					}
				}

				if voteNum > total/2 && rf.status == Candidate {
					DPrintf("%s get %d approve vote and start to be leader\n", rf.Self(), voteNum)
					rf.initLeader()
					rf.persist()
					rf.mu.Unlock()
					return
				}

				rf.mu.Unlock()
				if ok && reply.Success {
					return
				}
				time.Sleep(time.Millisecond * 50)
			}
		}(i)
	}
}

// doHeartbeat 向其他节点发送心跳包
// 主节点在同步日志成功后不会显示地发送提交请求，这将通过心跳包发送，以减少网络传输
func (rf *Raft) doHeartbeat() {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	prevLogIndex := rf.getLastLogIndex()
	prevLogTerm := rf.LastIncludedTerm
	if prevLogIndex > rf.LastIncludedIndex {
		prevLogTerm = rf.logs[rf.getArrayIndexByLogIndex(prevLogIndex)].Term
	}
	args := AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderId:     rf.me,
		Entries:      make([]LogEntry, 0),
		PrevLogIndex: prevLogIndex,
		PrevLogTerm:  prevLogTerm,
	}

	if rf.status != Leader {
		return
	}

	//DPrintf("leader %s start heartbeat\n", rf.self())
	for i := range rf.peers {
		if i == rf.me {
			continue
		}
		go func(id int) {
			// 对对方已经和 Leader 同步的日志进行提交请求
			// 这里不需要锁，即使是过时的数据也不会导致出错，顶多就是同步慢了一点而已
			args.LeaderCommit = int(math.Min(float64(rf.commitIndex), float64(rf.matchIndex[id])))
			reply := AppendEntriesReply{}
			ok := rf.sendAppendEntries(id, &args, &reply)

			rf.mu.Lock()
			defer rf.mu.Unlock()
			if ok && reply.Term > rf.currentTerm {
				rf.status = Follower
				rf.currentTerm = reply.Term
				rf.voteFor = -1
				rf.flushTime = time.Now().UnixMilli()
				rf.persist()
				return
			}
		}(i)
	}
}

// initLeader 方法初始化领导人状态，并发送空日志，其是不安全的，调用此方法必须持有锁
func (rf *Raft) initLeader() {
	defer rf.persist()
	rf.status = Leader
	rf.voteFor = rf.me

	rf.matchIndex = make([]int, len(rf.peers))
	rf.nextIndex = make([]int, len(rf.peers))

	for i := range rf.nextIndex {
		rf.nextIndex[i] = rf.getLastLogIndex() + 1
	}

	// 同步一下日志
	rf.syncLog()
}

func (rf *Raft) applyTicker() {
	for rf.killed() == false {
		rf.mu.Lock()

		for rf.lastApplied >= rf.commitIndex || rf.lastApplied == rf.getLastLogIndex() {
			rf.applyCond.Wait()
		}
		//DPrintf("%s apply message, current lastApplied %d, commitIndex %d, log length %d", rf.self(), rf.lastApplied, rf.commitIndex, len(rf.logs))

		applyMessages := make([]ApplyMsg, 0)

		for i := rf.lastApplied + 1; i <= rf.commitIndex; i++ {
			rf.lastApplied++
			if rf.getArrayIndexByLogIndex(i) < 0 {
				DPrintf("%s apply log, now lastApplied is %d, commitIndex is %d, i is %d, "+
					"lastIncludeIndex is %d, self log %+v\n", rf.Self(), rf.lastApplied, rf.commitIndex, i, rf.LastIncludedIndex, rf.logs)
			}
			applyMsg := ApplyMsg{
				CommandValid: true,
				Command:      rf.logs[rf.getArrayIndexByLogIndex(i)].Command,
				CommandIndex: i,
			}
			applyMessages = append(applyMessages, applyMsg)
		}
		DPrintf("%s apply log, now lastApplied is %d, commitIndex is %d apply logs %+v\n", rf.Self(), rf.lastApplied, rf.commitIndex, applyMessages)
		DPrintf("%s continue, lastinclud is %d, log is %+v\n", rf.Self(), rf.LastIncludedIndex, rf.logs)
		rf.mu.Unlock()

		for i := range applyMessages {
			*rf.applyCh <- applyMessages[i]
		}
		rf.persist()
	}
}

func (rf *Raft) Self() string {
	return fmt.Sprintf("server %d-%d-%d", rf.me, rf.status, rf.currentTerm)
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	defer rf.persist()

	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.currentTerm = 0
	rf.voteFor = -1
	rf.logs = make([]LogEntry, 0, 0)
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.LastIncludedIndex = 0
	rf.LastIncludedTerm = 0
	rf.matchIndex = make([]int, len(peers))
	rf.nextIndex = make([]int, len(peers))
	rf.status = Follower
	rf.flushTime = -1
	rf.applyCh = &applyCh
	rf.applyCond = sync.NewCond(&rf.mu)

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.electionTicker()
	go rf.heartbeatTicker()
	go rf.applyTicker()

	// init and apply
	rf.applyCond.Signal()

	return rf
}

// for lab3
func (rf *Raft) RaftStateSize() int {
	return rf.persister.RaftStateSize()
}
